package cn.tedu.process.flink;

import org.apache.flink.api.common.functions.AggregateFunction;
import cn.tedu.process.param.ChargingProgressParam;

/**
 * createAccumulator - 初始化一个新的累加器，这里是TemperatureAccumulator，它将存储温度的总和和温度值的数量。
 *
 * add - 每次来新的ChargingProgressParam时，更新累加器，增加温度值，并增加计数。
 *
 * getResult - 当时间窗口关闭时，从累加器中获取结果，即计算平均温度。
 *
 * merge - 在Flink需要合并两个分区的结果时使用，这在分布式环境中非常常见。
 */
public class TemperatureAggregator implements AggregateFunction<ChargingProgressParam, TemperatureAccumulator, Double> {

    // 创建新的累加器
    @Override
    public TemperatureAccumulator createAccumulator() {
        return new TemperatureAccumulator();
    }

    // 添加一个元素到累加器，更新温度总和和计数
    @Override
    public TemperatureAccumulator add(ChargingProgressParam value, TemperatureAccumulator acc) {
        acc.totalTemperature += value.getTemperature();
        acc.count += 1;
        return acc;
    }

    // 从累加器获取结果，计算平均温度
    @Override
    public Double getResult(TemperatureAccumulator acc) {
        if (acc.count == 0) {
            return 0.0; // 防止除以零
        } else {
            return acc.totalTemperature / acc.count;
        }
    }

    // 合并两个累加器，用于在并行环境下的聚合
    @Override
    public TemperatureAccumulator merge(TemperatureAccumulator a, TemperatureAccumulator b) {
        a.totalTemperature += b.totalTemperature;
        a.count += b.count;
        return a;
    }
}

// 累加器用于存储温度总和和计数
class TemperatureAccumulator {
    double totalTemperature = 0;
    int count = 0;
}
